package com.atguigu.tingshu.order.service.impl;

import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.order.service.OrderInfoService;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author: atguigu
 * @create: 2024-03-11 11:34
 */
@Slf4j
@Component
public class DelayMessageListener {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private OrderInfoService orderInfoService;

    /**
     * 监听延迟关闭订单消息
     * 开启一个单一线程线程池负责监听延迟队列中消息
     */
    @PostConstruct
    public void cancelOrderDelayMsg() {
        log.info("开启监听延迟关单消息....");
        //1.创建线程池 单一线程
        ExecutorService executor = Executors.newSingleThreadExecutor();
        //2.提交任务
        executor.submit(() -> {
            //2.1 获取阻塞队列
            RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(KafkaConstant.QUEUE_ORDER_CANCEL);
            //2.1 开启采用循环方式获取阻塞队列中消息
            while (true) {
                String orderId = blockingQueue.poll();
                if (StringUtils.isNotBlank(orderId)) {
                    log.info("【订单服务】监听到延迟关单消息：{}", orderId);
                    orderInfoService.orderCancel(Long.valueOf(orderId));
                }
            }
        });
    }
}
